package drds.data_migrate.util;

import drds.common.$;
import drds.common.Author;
import drds.data_migrate.Authors;
import drds.data_migrate.TableSplits;
import drds.data_migrate.vo.ConnectionInfo;
import drds.data_migrate.vo.JobConfig;
import drds.data_migrate.vo.TaskConfig;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

@Author(name = Authors.LI_YANG)
public final class ReaderSplitUtil
{


    public static List<TaskConfig> doSplit(JobConfig jobConfig, int adviceNumber) throws SQLException
    {

        int adviceTaskSize = -1;
        // adviceNumber这里是channel数量大小, 即datax并发task数量
        // eachTableShouldSplittedNumber是单表应该切分的份数, 向上取整可能和adviceNumber没有比例关系了已经
        adviceTaskSize = calculateEachTableShouldSplittedNumber(adviceNumber, jobConfig.getTableNumberMark());

        String columnNamesString = jobConfig.getColumnNamesString();
        String where = jobConfig.getWhere();

        List<ConnectionInfo> connectionInfoList = jobConfig.getConnectionInfoList();

        List<TaskConfig> splittedTaskConfigs = new ArrayList<TaskConfig>();

        for (int i = 0, size = connectionInfoList.size(); i < size; i++)
        {
            TaskConfig taskConfig = new TaskConfig();

            ConnectionInfo connectionInfo = connectionInfoList.get(i);
            String jdbcUrl = connectionInfo.getJdbcUrl();
            taskConfig.setJdbcUrl(jdbcUrl);

            // 抽取 jdbcUrl 中的 ip/port 进行资源使用的打标，以提供给 core 做有意义的 shuffle 操作
            taskConfig.setLoadBalanceResourceMark(jdbcUrl);//ip


            // 已在之前进行了扩展和`处理，可以直接使用
            List<String> tableNameList = connectionInfo.getTableNameList();
            String splitColumnName = jobConfig.getSplitColumnName();

            //最终切分份数不一定等于 eachTableShouldSplittedNumber
            boolean needSplitTable = adviceTaskSize > 1 && $.isNotNullAndNotEmpty(splitColumnName);
            if (needSplitTable)
            {
                if (tableNameList.size() == 1)
                {
                    //原来:如果是单表的，主键切分num=num*2+1
                    // splitPk is null这类的情况的数据量本身就比真实数据量少很多, 和channel大小比率关系时，不建议考虑
                    //eachTableShouldSplittedNumber = eachTableShouldSplittedNumber * 2 + 1;// 不应该加1导致长尾

                    //考虑其他比率数字?(splitPk is null, 忽略此长尾)
                    adviceTaskSize = adviceTaskSize * 5;
                }
                // 尝试对每个表，切分为eachTableShouldSplittedNumber 份
                for (String tableName : tableNameList)
                {
                    //splittedTaskConfig = sliceTaskConfig;
                    taskConfig.setTableName(tableName);

                    List<TaskConfig> splittedSlices = TableSplits.split(jobConfig, adviceTaskSize);

                    splittedTaskConfigs.addAll(splittedSlices);
                }
            } else
            {
                for (String tableName : tableNameList)
                {
                    //splittedTaskConfig = sliceTaskConfig;
                    taskConfig.setTableName(tableName);
                    String queryColumn = columnNamesString;
                    taskConfig.setQuerySql(TableSplits.buildQuerySql(tableName, queryColumn, where));
                    splittedTaskConfigs.add(taskConfig);
                }
            }


        }

        return splittedTaskConfigs;
    }

    public static JobConfig doPreCheckSplit(JobConfig jobConfig)
    {


        String columnNamesString = jobConfig.getColumnNamesString();
        String where = jobConfig.getWhere();
        //
        String splitColumnName = jobConfig.getSplitColumnName();

        List<ConnectionInfo> connectionInfoList = jobConfig.getConnectionInfoList();

        for (int i = 0, size = connectionInfoList.size(); i < size; i++)
        {
            ConnectionInfo connectionInfo = connectionInfoList.get(i);
            List<String> querySqlList = new ArrayList<String>();
            List<String> splitColumnRangeSqlList = new ArrayList<String>();

            {
                // 已在之前进行了扩展和`处理，可以直接使用
                List<String> tableNameList = connectionInfo.getTableNameList();

                for (String tableName : tableNameList)
                {
                    querySqlList.add(TableSplits.buildQuerySql(tableName, columnNamesString, where));
                    if (splitColumnName != null && splitColumnName.length() > 0)
                    {
                        splitColumnRangeSqlList.add(TableSplits.generateSplitColumnRangeSql(tableName, splitColumnName.trim(), where));
                    }
                }
                if ($.isHasData(splitColumnRangeSqlList))
                {
                    connectionInfo.setSplitColumnRangeSqlList(splitColumnRangeSqlList);
                }
                connectionInfo.setQuerySqlList(querySqlList);

            }
        }
        return jobConfig;
    }

    private static int calculateEachTableShouldSplittedNumber(int adviceNumber,
                                                              int tableNumber)
    {
        double tempNum = 1.0 * adviceNumber / tableNumber;

        return (int) Math.ceil(tempNum);
    }

}
